using System;
using System.Collections.Generic;
using System.Threading;

namespace Framework.Library.Core.Utils.Threading
{
    public delegate void ConsumptionCycle<T>(IEnumerable<T> consumptionLine, object state);

    public class Consumer<T>
    {
        private readonly Producer<T> _producer;
        private ConsumptionCycle<T> _consumptionCycle;

        public Consumer(Producer<T> producer)
        {
            _producer = producer;
        }

        public Producer<T> Producer
        {
            get { return _producer; }
        }

        public IAsyncResult BeginConsume(ConsumptionCycle<T> consumptionCycle, object obj, AsyncCallback callback,
                                         object state)
        {
            if (consumptionCycle == null)
                throw new ArgumentNullException("consumptionCycle", "ConsumptionCycle delegate cannot be null");

            var cancelableResult = new CancelableAsyncResult();

            // wrap user callback delegate to pass cancelable result
            AsyncCallback fixedCallback = null;
            if (callback != null)
            {
                fixedCallback = delegate { callback(cancelableResult); };
            }

            _consumptionCycle = consumptionCycle;
            cancelableResult.Result =
                _consumptionCycle.BeginInvoke(_producer.GetConsumerChannel(cancelableResult, Timeout.Infinite), obj,
                                              fixedCallback, state);

            return cancelableResult;
        }

        public void EndConsume(IAsyncResult result)
        {
            var cancelableResult = (CancelableAsyncResult) result;

            _consumptionCycle.EndInvoke(cancelableResult.Result);
            cancelableResult.CancelSignal.Close();
        }

        public void Consume(ConsumptionCycle<T> consumptionCycle, object obj)
        {
            Consume(consumptionCycle, obj, Timeout.Infinite);
        }

        public void Consume(ConsumptionCycle<T> consumptionCycle, object obj, int millisecondsTimeout)
        {
            if (consumptionCycle == null)
                throw new ArgumentNullException("consumptionCycle", "ConsumptionCycle delegate cannot be null");

            consumptionCycle(_producer.GetConsumerChannel(null, millisecondsTimeout), obj);
        }
    }
}